package com.ywu.mq.sdk.consumer;

import com.ywu.boot.context.YwuContext;
import com.ywu.boot.context.YwuContextHolder;
import com.ywu.boot.context.YwuContextTenant;
import com.ywu.common.gson.GsonUtils;
import com.ywu.common.result.CommonResult;
import com.ywu.common.utils.TimerIdGenerateUtil;
import com.ywu.mq.sdk.config.YwuKafkaSdkConfig;
import com.ywu.mq.spi.dto.YwuKafkaDto;
import com.ywu.mq.spi.entity.YwuKafkaConsumerLogEntity;
import com.ywu.mq.spi.spi.YwuKafkaSpi;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @ClassName YwuKafkaConsumer
 * @Description TODO
 * @Author GroundDemo
 * @Date 2025/3/28 15:52
 * @Version 1.0
 **/
@Component
@Slf4j
public abstract class YwuKafkaConsumer {

    private YwuKafkaSdkConfig sdkConfig;

    @Autowired
    private YwuKafkaSpi kafkaSpi;

    /**
    * @Author GroundDemo
    * @Description 初始化，构造配置类对象，执行消费监听
    * @Date 18:03 2025/3/28
    * @Param [config]
    * @return void
    **/
    public void init(YwuKafkaSdkConfig config) {
        this.sdkConfig = config;
        consumer();
    }

    /**
    * @Author GroundDemo
    * @Description 消费方法
    * @Date 16:24 2025/3/28
    * @Param [key, value]
    * @return void
    **/
    public abstract void doCustomConsumer(String key, String value);

    /**
     * @Author GroundDemo
     * @Description 消费消息 - 服务启动时就开始消费
     * @Date 16:13 2025/3/28
     * @Param []
     * @return void
     **/
    private void consumer() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, sdkConfig.getBootstrapService());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, sdkConfig.getKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, sdkConfig.getValueDeserializer());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, sdkConfig.getGroupId());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, sdkConfig.getOffsetReset());
        // 查询当前主题是否有权限消费
        YwuKafkaDto ywuKafkaDto = new YwuKafkaDto();
        ywuKafkaDto.setTopicName(sdkConfig.getTopicName());
        ywuKafkaDto.setTopicAppId(sdkConfig.getTopicAppId());
        ywuKafkaDto.setAppId(sdkConfig.getAppId());
        ywuKafkaDto.setAppKey(sdkConfig.getAppKey());
        CommonResult<Object> result = kafkaSpi.checkConsumerPermission(ywuKafkaDto);
        if (Objects.isNull(result) || !result.getCode().equals("200")) {
            log.error("topic name permission check spi error {}", GsonUtils.serialize(result));
            throw new RuntimeException("topic name permission check spi error");
        }
        String appId = sdkConfig.getTopicAppId();
        String replaceAppId = appId.replace(".", "_").replace("-", "_");
        String topicName = sdkConfig.getTopicName();
        String topic = replaceAppId + "-" + topicName;
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(sdkConfig.getConsumerNum());
        // 启动多个线程
        for (int i = 0; i < sdkConfig.getConsumerNum(); i++) {
            executorService.submit(new Thread(() -> {
                // 创建消费者实例
                KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
                // 订阅主题
                consumer.subscribe(Collections.singletonList(topic));
                try {
                    while (true) {
                        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
                        for (ConsumerRecord<String, String> record : records) {
                            doConsumer(record);
                        }
                    }
                } finally {
                    // 关闭消费者
                    consumer.close();
                }
            }));
        }
    }

    /**
    * @Author GroundDemo
    * @Description 消费数据
    * @Date 16:23 2025/3/28
    * @Param [record]
    * @return void
    **/
    private void doConsumer(ConsumerRecord<String, String> record) {
        // 恢复value值
        String value = record.value();
        String[] split = value.split("-ywuMessageId-");
        String messageId = split[0];
        String realValue = split[1];
        // 记录消费日志
        YwuKafkaConsumerLogEntity ywuKafkaConsumerLogEntity = new YwuKafkaConsumerLogEntity();
        ywuKafkaConsumerLogEntity.setId(TimerIdGenerateUtil.nextId());
        if (StringUtils.isEmpty(sdkConfig.getAppId())) {
            ywuKafkaConsumerLogEntity.setConsumerApp(sdkConfig.getTenantId());
        } else {
            ywuKafkaConsumerLogEntity.setConsumerApp(sdkConfig.getAppId());
        }
        ywuKafkaConsumerLogEntity.setConsumerStatus(YwuKafkaConsumerLogEntity.CONSUMING);
        ywuKafkaConsumerLogEntity.setMessageId(messageId);
        ywuKafkaConsumerLogEntity.setBeginTime(new Date());
        ywuKafkaConsumerLogEntity.setGroupId(sdkConfig.getGroupId());
        kafkaSpi.createConsumerLog(ywuKafkaConsumerLogEntity);
        // 执行消费方法
        doCustomConsumer(record.key(), realValue);
        ywuKafkaConsumerLogEntity.setEndTime(new Date());
        ywuKafkaConsumerLogEntity.setConsumerStatus(YwuKafkaConsumerLogEntity.CONSUMED);
        YwuContext currentContext = YwuContextHolder.getCurrentContext();
        // 异步调用spi记录日志
        new Thread(() -> {
            YwuContextHolder.setCurrentContext(currentContext);
            kafkaSpi.updateConsumerLog(ywuKafkaConsumerLogEntity);
        }).start();
    }

}
